/*******************************************************************************
 *
 * DSM - Distributed Search Manager
 * Developed by Milspec Research International Pty Ltd
 * 
 * $Author: gsingh $
 * $Revision: 1.2 $
 * $Date: 2004/06/01 02:46:15 $
 *
 * (c)Copyright 2004
 *    education.au limited 
 *    DEST
 *    All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without 
 * modification, are permitted provided that the following conditions are met:
 *
 * Redistributions of source code must retain the above copyright notice, this 
 * list of conditions and the following disclaimer.
 *
 * Redistributions in binary form must reproduce the above copyright notice, 
 * this list of conditions and the following disclaimer in the documentation 
 * and/or other materials provided with the distribution.
 *
 * Neither the names education.au limited, DEST nor the names of its 
 * contributors may be used to endorse or promote products derived from this 
 * software without specific prior written permission.
 *
 * To the extent permitted by law, the copyright owners of this software and 
 * its contributors:
 *
 *  (i) exclude all warranties in relation to the software; and
 * (ii) exclude liability for all losses, costs, expenses and damages arising 
 *      in any way from the use of the software whether arising from or in 
 *      relation to breach of contract, negligence or any other tort, in equity
 *      or otherwise. If the software is in breach of a warranty which is 
 *      implied by law, the copyright owners and contributors liability is
 *      limited to the replacement of the software.
 *
 *******************************************************************************
 */

package au.edu.educationau.opensource.dsm.threadpool;

/**
 * Object Queue - First in First Out policy This is based on the Thread-safe
 * ObjectFIFO class written by Paul Hyde
 * 
 */
public class ObjectFIFO extends java.lang.Object {
	private Object[] queue;
	private int capacity;
	private int size;
	private int head;
	private int tail;

	/**
	 * Constructor
	 * 
	 * @param cap
	 *            the capacity of this store
	 */
	public ObjectFIFO(int cap) {
		capacity = (cap > 0) ? cap : 1; // at least 1
		queue = new Object[capacity];
		head = 0;
		tail = 0;
		size = 0;
	}

	/** Returns the capacity of the store */
	public int getCapacity() {
		return capacity;
	}

	/** Returns the number of objects in this store */
	public synchronized int getSize() {
		return size;
	}

	/** Checks if the store is empty */
	public synchronized boolean isEmpty() {
		return (size == 0);
	}

	/** Checks if the store is full */
	public synchronized boolean isFull() {
		return (size == capacity);
	}

	/**
	 * Adds a new object to the store
	 * 
	 * @param obj
	 * @exception InterruptedException
	 */
	public synchronized void add(Object obj) throws InterruptedException {

		waitWhileFull();

		queue[head] = obj;
		head = (head + 1) % capacity;
		size++;

		notifyAll(); // let any waiting threads know about change
	}

	/**
	 * Adds a list to the existing list
	 * 
	 * @param list
	 * @exception InterruptedException
	 */
	public synchronized void addEach(Object[] list) throws InterruptedException {

		for (int i = 0; i < list.length; i++) {
			add(list[i]);
		}
	}

	/**
	 * Gets the next object
	 * 
	 * @exception InterruptedException
	 */
	public synchronized Object remove() throws InterruptedException {

		waitWhileEmpty();

		Object obj = queue[tail];

		// don't block GC by keeping unnecessary reference
		queue[tail] = null;

		tail = (tail + 1) % capacity;
		size--;

		notifyAll(); // let any waiting threads know about change

		return obj;
	}

	/**
	 * Removes all the objects
	 * 
	 * @exception InterruptedException
	 */
	public synchronized Object[] removeAll() throws InterruptedException {

		Object[] list = new Object[size]; // use the current size

		for (int i = 0; i < list.length; i++) {
			list[i] = remove();
		}

		// if FIFO was empty, a zero-length array is returned
		return list;
	}

	/**
	 * removes all of them until all empty
	 * 
	 * @exception InterruptedException
	 */
	public synchronized Object[] removeAtLeastOne() throws InterruptedException {

		waitWhileEmpty(); // wait for a least one to be in FIFO
		return removeAll();
	}

	/**
	 * Waits for msTimeout until the queue is empty
	 * 
	 * @param msTimeout
	 * @exception InterruptedException
	 */
	public synchronized boolean waitUntilEmpty(long msTimeout) throws InterruptedException {

		if (msTimeout == 0L) {
			waitUntilEmpty(); // use other method
			return true;
		}

		// wait only for the specified amount of time
		long endTime = System.currentTimeMillis() + msTimeout;
		long msRemaining = msTimeout;

		while (!isEmpty() && (msRemaining > 0L)) {
			wait(msRemaining);
			msRemaining = endTime - System.currentTimeMillis();
		}

		// May have timed out, or may have met condition,
		// calc return value.
		return isEmpty();
	}

	/**
	 * Waits until queue is empty
	 * 
	 * @exception InterruptedException
	 */
	public synchronized void waitUntilEmpty() throws InterruptedException {

		while (!isEmpty()) {
			wait();
		}
	}

	/**
	 * Wait while the queue is empty
	 * 
	 * @exception InterruptedException
	 */
	public synchronized void waitWhileEmpty() throws InterruptedException {

		while (isEmpty()) {
			wait();
		}
	}

	/**
	 * Wait until the queue is full
	 * 
	 * @exception InterruptedException
	 */
	public synchronized void waitUntilFull() throws InterruptedException {

		while (!isFull()) {
			wait();
		}
	}

	/**
	 * Wait while the queue is full
	 * 
	 * @exception InterruptedException
	 */
	public synchronized void waitWhileFull() throws InterruptedException {

		while (isFull()) {
			wait();
		}
	}
}
